package com.gitee.app;

import com.gitee.poto.Article;
import com.google.gson.Gson;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;

public class AppMain {
    private static final String tableName = "hbase_es_article";
    private static final String familyName = "f1";
    private static final String title = "title";
    private static final String from = "from";
    private static final String times = "times";
    private static final String readCounts = "readCounts";
    private static final String content = "content";

    public static void main(String[] args) throws IOException {
        /*//使用java代码解析excel表格
        List<Article> exceInfo = ExcelUtil.getExceInfo();

        //将集合当中的数据，保存到es当中去
        TransportClient client = getEsClient();
        save2Es(exceInfo, client);
        Table table = getTable();
        //循环遍历我们的数据，将我们的数据装到List<Put>
        saveToHbase(exceInfo, table);*/

        //通过对 "机器人" 关键字的搜索找到对应的rowkey并返回所以rowkey的一个集合
        TransportClient esClient = getEsClient();
        List<String>  getAllKeyWord =  getByKeyWord(esClient,"机器人");
        Table table = getTable();
        //创建一个Article的集合返回给前端页面
        ArrayList<Article> articles = new ArrayList<>();
        for (String rowKey : getAllKeyWord) {
            //拿一个rowkey作为验证,查看内容是否包含 "机器人" 关键字
            Get get = new Get(rowKey.getBytes());
            Result result = table.get(get);
            Cell[] cells = result.rawCells();
            //cells[0]是内容
            //cells[1]是作者
            //cells[2]是访问量
            //cells[3]是时间
            //cells[4]是标题
            String content = Bytes.toString(cells[0].getValue());
            String from = Bytes.toString(cells[1].getValue());
            String readCounts = Bytes.toString(cells[2].getValue());
            String times = Bytes.toString(cells[3].getValue());
            String title = Bytes.toString(cells[4].getValue());
            Article article = new Article();
            article.setContent(content);
            article.setFrom(from);
            article.setReadCounts(readCounts);
            article.setTimes(times);
            article.setTitle(title);
            articles.add(article);
        }

        //最后将articles返回给前端页面就行了
//        articles
    }

    private static List<String> getByKeyWord(TransportClient esClient, String keyWord) {
        ArrayList<String> strings = new ArrayList<>();
        SearchResponse searchResponse = esClient.prepareSearch("articles")
                .setTypes("article")
                .setQuery(QueryBuilders.termQuery("title", keyWord))
                .get();
        SearchHits hits = searchResponse.getHits();
        for (SearchHit hit : hits) {
            //获取我们数据的系统id
            String id = hit.getId();
            strings.add(id);
        }
        return strings;
    }

    /**
     * 将数据保存到hbase
     *
     * @param exceInfo
     * @param table
     */
    private static void saveToHbase(List<Article> exceInfo, Table table) throws IOException {
        System.out.println(exceInfo.size());
        //获取开始时间
        long startTime = System.currentTimeMillis();
        ArrayList<Put> puts = new ArrayList<>();
        for (Article article : exceInfo) {
            //打印标题
            System.out.println(article.getTitle());
            Put put = new Put(Bytes.toBytes(article.getId()));
            if (article.getTitle() != null && article.getTitle() != "") {
                put.addColumn(familyName.getBytes(), title.getBytes(), article.getTitle().getBytes());
                put.addColumn(familyName.getBytes(), from.getBytes(), article.getFrom().getBytes());
                put.addColumn(familyName.getBytes(), times.getBytes(), article.getTimes().getBytes());
                put.addColumn(familyName.getBytes(), readCounts.getBytes(), article.getReadCounts().getBytes());
                put.addColumn(familyName.getBytes(), content.getBytes(), article.getContent().getBytes());
                puts.add(put);
            }
        }
        table.put(puts);
        //获取结束时间
        long endTime = System.currentTimeMillis();
        //打印存入的时间间隔
        System.out.println((endTime-startTime)/1000);
        table.close();
    }

    /**
     * 建立hbase连接和创建表名
     *
     * @return
     */
    private static Table getTable() throws IOException {
        //获取hbase连接
        Configuration configuration = HBaseConfiguration.create();
        //zookeeper的集群地址
        configuration.set("hbase.zookeeper.quorum", "node01:2181,node02:2181,node03:2181");
        Connection connection = ConnectionFactory.createConnection(configuration);
        //获取当前用户,为等会儿判断表是否存在
        Admin admin = connection.getAdmin();
        //设置表名
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(familyName);
        hTableDescriptor.addFamily(hColumnDescriptor);
        //如果表不存在就创建
        if (!admin.tableExists(TableName.valueOf(tableName))) {
            admin.createTable(hTableDescriptor);
        }
        return connection.getTable(TableName.valueOf(tableName));
    }

    /**
     * 保存数据到es
     *
     * @param exceInfo
     * @param client
     */
    private static void save2Es(List<Article> exceInfo, TransportClient client) {
        //通过批量添加将我们的数据存储到es中
        BulkRequestBuilder bulk = client.prepareBulk();
        //循环遍历集合添加到bulk中
        for (Article article : exceInfo) {
            IndexRequestBuilder indexRequestBuilder = client.prepareIndex("articles", "article", article.getId());
            Gson gson = new Gson();
            String jsonStr = gson.toJson(article);
            indexRequestBuilder.setSource(jsonStr, XContentType.JSON);
            bulk.add(indexRequestBuilder);
        }
        //触发数据真正保存到es中去
        bulk.get();
        client.close();
    }

    /**
     * 建立一个es连接
     *
     * @return
     */
    private static TransportClient getEsClient() throws UnknownHostException {
        //put中内容是由es中配置决定的
        // /export/servers/es/elasticsearch-6.7.0/config/elasticsearch.yml 中的集群名字 cluster.name: myes
        Settings settings = Settings.builder().put("cluster.name", "myes").build();
        TransportClient client = new PreBuiltTransportClient(settings)
                .addTransportAddresses(new TransportAddress(InetAddress.getByName("node01"), 9300),
                        new TransportAddress(InetAddress.getByName("node02"), 9300));
        return client;
    }

}
